package com.zt.bigdata.spark.dataalgorithms.chapter05

import com.zt.bigdata.template.spark.BasicTemplate

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
  *
  */
class RelativeFrequency extends BasicTemplate[Parameter] {
  override def process(parameter: Parameter): Unit = {


    val spark = buildSparkSession(parameter)
    val inputFile = spark.sparkContext.textFile(parameter.inputFile)
    val broadcastWindow = spark.sparkContext.broadcast(5)
    val pairs = inputFile.flatMap {
      line =>
        val tokens = line.split("\\s")
        for {
          i <- 0 until tokens.length
          start = if (i - broadcastWindow.value < 0) 0 else i - broadcastWindow.value
          end = if (i + broadcastWindow.value >= tokens.length) tokens.length - 1 else i + broadcastWindow.value
          j <- start to end if (j != i)
        } yield (tokens(i), (tokens(j), 1))
    }
    pairs.collect().foreach(println)

    val totalByKey = pairs.map(x => (x._1, x._2._2)).reduceByKey(_ + _)
    totalByKey.collect().foreach(println)

    val grouped = pairs.groupByKey()
    val uniquePairs2 = grouped.flatMapValues(
      x =>
        x.groupBy(x => x._1)
          .mapValues(y => y.unzip._2.sum)
    )
    val uniquePairs = grouped.flatMapValues {
      iter =>
        val map = mutable.HashMap.empty[String, Int]
        val list = ArrayBuffer.empty[(String, Int)]
        iter.foreach {
          value =>

            /**
              * (w1,(w1,1))
              * (w1,(w2,1))    (w1,(w3,1))
              * (w1,(w1,1)) => (w1,(w2,2))
              * (w1,(w2,1))    (w1,(w1,2))
              * (w1,(w3,1))
              */
            var total = value._2
            if (map.contains(value._1))
              total = total + map.get(value._1).get
            map.+=(value._1 -> total)
        }
        map.foreach(x => list.+=((x._1, x._2)))
        list
    }
    uniquePairs.collect().foreach(println)

    val joined = uniquePairs.join(totalByKey)
    joined.collect().foreach(println)

    joined.map {
      x =>
        ((x._1, x._2._1._1), (x._2._1._2.toDouble / x._2._2.toDouble).formatted("%.5f"))
    }.collect().foreach(println)
  }
}

object T {
  def main(args: Array[String]): Unit = {
    val pair = List(("a", 1), ("b", 2), ("c", 3), ("d", 4))
    println(pair.unzip._1)
    println(pair.unzip._2)
  }
}